package site.teamo.biu.flink;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.io.resource.ResourceUtil;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.codehaus.janino.ClassBodyEvaluator;
import site.teamo.biu.flink.sdk.job.JobContext;
import site.teamo.biu.flink.sdk.job.JobInfo;
import site.teamo.biu.flink.sdk.job.JobUtil;
import site.teamo.biu.flink.sdk.job.Pipeline;

import java.util.List;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author haocongshun
 * @date 2023/07/25 14:13:25
 */
@Slf4j
public class BiuFlinkCoreApplication {
    public static void main(String[] args) throws Exception {
        if (args.length < 1) {
            log.error("参数缺少配置文件路径");
            return;
        }
        String jobConfigFilePath = args[0];
        JobInfo job = JSON.parseObject(FileUtil.readUtf8String(jobConfigFilePath), JobInfo.class);
        List<Pipeline> pipelines = JobUtil.toPipelines(job);
        StringBuilder sb = new StringBuilder();
        for (Pipeline pipeline : pipelines) {
            sb.append(pipeline.toCode());
        }
        String code = String.format(ResourceUtil.readUtf8Str("template/JobApplicationTemplate.java"), sb);
//        System.out.println(code);
        // 使用Janino编译器编译Java源代码
        ClassBodyEvaluator classBodyEvaluator = new ClassBodyEvaluator();
        classBodyEvaluator.cook(code);
        // 加载并实例化生成的类
        Class<?> templateClass = classBodyEvaluator.getClazz();
        Object template = templateClass.getDeclaredConstructor().newInstance();
        // 调用生成的类的方法
        try (StreamExecutionEnvironment env = createEnv()) {
            templateClass.getDeclaredMethod("fillJob", StreamExecutionEnvironment.class, JobContext.class)
                    .invoke(template, env, new JobContext(job));
            env.execute(job.getName());
        }
    }

    private static StreamExecutionEnvironment createEnv() {
        return StreamExecutionEnvironment.getExecutionEnvironment();
    }
}
